package org.example.ItemCF.src.recommend;

import java.io.IOException;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.example.ItemCF.src.hdfs.HdfsDAO;


/**
 * 第二步：将用户向量转换为共现关系。
 * 此步骤的目的是通过映射和减少操作，从用户对项目的偏好向量中生成项目对的共现频率。
 */
public class Step2Driver {

	/**
	 * 执行第二步的共现计算作业。
	 * 此步骤基于用户向量计算用户之间的共现关系。
	 *
	 * @param path 包含输入和输出路径的映射，其中"Step2Input"为输入路径，"Step2Output"为输出路径。
	 * @throws IOException 在进行文件读写时可能会抛出的异常。
	 * @throws ClassNotFoundException 在加载类时可能会抛出的异常。
	 * @throws InterruptedException 在作业执行过程中可能会抛出的中断异常。
	 */
	public static void run(Map<String, String> path) throws IOException,
	        ClassNotFoundException, InterruptedException {
	    Configuration conf = new Configuration();
	    String input = path.get("Step2Input"); // 获取输入路径
	    String output = path.get("Step2Output"); // 获取输出路径
	    // 初始化HDFS操作
	    HdfsDAO hdfs = new HdfsDAO(Recommend.HDFS, conf);
	    hdfs.rmr(output); // 清理输出目录，确保作业开始前输出目录为空
	    // 配置作业参数
	    Job job = Job.getInstance(conf, "RecommendStep2");
	    job.setJarByClass(Step2Driver.class); // 设置代表作业的JAR文件
	    job.setMapOutputKeyClass(Text.class); // 设置映射器输出键的类型
	    job.setMapOutputValueClass(IntWritable.class); // 设置映射器输出值的类型
	    job.setOutputKeyClass(IntWritable.class); // 设置减少器输出键的类型
	    job.setOutputValueClass(Text.class); // 设置减少器输出值的类型
	    // 设置映射器和减少器的类
	    job.setMapperClass(Step2UserVectorToCooccurrenceMapper.class);
	    job.setReducerClass(Step2UserVectorToConoccurrenceReducer.class);
	    // 设置作业的输入和输出路径
	    FileInputFormat.addInputPath(job, new Path(input)); // 输入路径设置
	    FileOutputFormat.setOutputPath(job, new Path(output)); // 输出路径设置
	    // 提交作业并等待其完成
	    if (!job.waitForCompletion(true))
	        return; // 如果作业未成功完成，则返回
	}
}
